package com.wu.gconduit.common.util

import io.searchbox.client.JestClientFactory
import io.searchbox.client.config.HttpClientConfig
import io.searchbox.core.{Bulk, Index}

/**
 * @author Wumeng
 * @Date 2022年03月04日 15:18
 */
object ESUtil {
  val esUrl = "http://hadoop104:9200"

  //创建客户端工厂
  val factory = new JestClientFactory
  val config = new HttpClientConfig.Builder(esUrl)
    .maxTotalConnection(100) //最多有多少个连接
    .connTimeout(10 * 1000) //连接es超时时间
    .readTimeout(10 * 1000) //读取数据的超时时间
    .multiThreaded(true) //
    .build()

  factory.setHttpClientConfig(config)

  def insertSingle(index: String, source: Object, id: String = null) = {
    ///从工厂获取一个客户端
    val client = factory.getObject
    //3、写入（单次、批次）
    val action: Index = new Index.Builder(source)
      .index(index)
      .`type`("_doc")
      //      .id("1")  //可选，如果没有设置，会自动生成
      .build()

    client.execute(action)
    //4、关闭客户端
    client.shutdownClient()
  }

  //批次插入
  def insertBulk(index: String, sources: Iterator[Any]): Unit = {
    val client = factory.getObject
    val bulk = new Bulk.Builder()
      .defaultIndex(index)
      .defaultType("_doc")
    //
    //添加多条元素,循环的方式获取
    /*sources.foreach(source =>{
      val action = new Index.Builder(source).build()
      bulk.addAction(action)
    })*/

    sources.foreach {
      case (id: String, data) =>
        val action = new Index.Builder(data).id(id).build()
        bulk.addAction(action)

      case data =>
        val action = new Index.Builder(data).build()
        bulk.addAction(action)
    }

    client.execute(bulk.build())
    //4、关闭客户端
    client.shutdownClient()

  }


  def main(args: Array[String]): Unit = {
    //    val data = User(11,"wwwwwwwwwwwwww222222222")
    //    insertSingle("users",data)

    val list = User(1, "aaa") :: User(2, "bbb") :: User(3, "ccc") :: Nil
    val list1 = (100, User(1, "aaa111")) :: (200, User(2, "bbb222")) :: (300, User(3, "ccc333")) :: Nil


    insertBulk("users", list1.toIterator)
  }

}

case class User(age: Int, name: String)
